package com.fanli.bigdata.mytest

import kafka.serializer.StringDecoder
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka.KafkaUtils

object WindowsDemo {
  def main (args: Array[String]): Unit = {
    var conf = new SparkConf().setAppName("DSDEMO").setMaster("local[*]")
    var sc = new SparkContext(conf)
    var ssc = new StreamingContext(sc, Seconds(10))

    val kafkaParams = Map[String, String]("metadata.broker.list" -> "115.159.45.213:9092")
    val topics = "test02"
    val topicsSet = topics.split(",").toSet
    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

    val lines = messages.map(_._2)
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1L)).reduceByKeyAndWindow((a:Long,b:Long) => (a + b),Seconds(30), Seconds(10))
    wordCounts.print()
    ssc.start()
    ssc.awaitTermination()
  }
}
